package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// 使用函数式编程风格的桥接channel模式
// 桥接函数(bridge)接收一个产生channels的channel，
// 将它们扁平化为单个channel，按顺序消费所有channels
func bridge(done <-chan struct{}, chanStream <-chan <-chan interface{}) <-chan interface{} {
	valStream := make(chan interface{})

	go func() {
		defer close(valStream)

		// 以函数式编程风格处理channels流
		for {
			var stream <-chan interface{}
			select {
			case maybeStream, ok := <-chanStream:
				if !ok {
					return
				}
				stream = maybeStream
			case <-done:
				return
			}

			// 将从每个内部channel收到的值转发到输出channel
			for val := range orDone(done, stream) {
				select {
				case valStream <- val:
				case <-done:
					return
				}
			}
		}
	}()

	return valStream
}

// orDone 是一个组合器，用于处理取消信号
func orDone(done <-chan struct{}, c <-chan interface{}) <-chan interface{} {
	valStream := make(chan interface{})

	go func() {
		defer close(valStream)
		for {
			select {
			case <-done:
				return
			case v, ok := <-c:
				if !ok {
					return
				}
				select {
				case valStream <- v:
				case <-done:
					return
				}
			}
		}
	}()

	return valStream
}

// 一个函数，返回一个channel，该channel每秒生成一个channel
// 每个生成的channel包含一个整数流
func genChannelStream(done <-chan struct{}) <-chan <-chan interface{} {
	chanStream := make(chan (<-chan interface{}))

	// 用纯函数式方式生成一系列channels
	go func() {
		defer close(chanStream)
		for i := 0; i < 10; i++ {
			// 对每个i创建一个新的stream channel
			stream := makeIntStream(done, i)

			// 发送这个stream
			select {
			case <-done:
				return
			case chanStream <- stream:
			}

			// 每秒生成一个新的stream
			time.Sleep(1 * time.Second)
		}
	}()

	return chanStream
}

// 生成一个包含整数序列的channel
// 使用seed作为起始值，发送seed到seed+4
func makeIntStream(done <-chan struct{}, seed int) <-chan interface{} {
	intStream := make(chan interface{})

	go func() {
		defer close(intStream)

		// 基础值
		base := seed * 10

		// 生成5个整数
		for i := 0; i < 5; i++ {
			n := base + i

			// 模拟一些处理时间
			delay := time.Duration(rand.Intn(500)) * time.Millisecond
			time.Sleep(delay)

			select {
			case <-done:
				return
			case intStream <- n:
				fmt.Printf("发送值: %d (延迟: %v)\n", n, delay)
			}
		}

		fmt.Printf("stream %d 已完成\n", seed)
	}()

	return intStream
}

// 使用map转换流中的值
func mapStream(done <-chan struct{}, in <-chan interface{}, fn func(interface{}) interface{}) <-chan interface{} {
	mapped := make(chan interface{})

	go func() {
		defer close(mapped)

		for val := range orDone(done, in) {
			select {
			case <-done:
				return
			case mapped <- fn(val):
			}
		}
	}()

	return mapped
}

// 使用filter过滤流中的值
func filterStream(done <-chan struct{}, in <-chan interface{}, fn func(interface{}) bool) <-chan interface{} {
	filtered := make(chan interface{})

	go func() {
		defer close(filtered)

		for val := range orDone(done, in) {
			if fn(val) {
				select {
				case <-done:
					return
				case filtered <- val:
				}
			}
		}
	}()

	return filtered
}

// 使用reduce对流中的值进行聚合
func reduceStream(done <-chan struct{}, in <-chan interface{}, fn func(acc, val interface{}) interface{}, initial interface{}) interface{} {
	result := initial

	for val := range orDone(done, in) {
		result = fn(result, val)
	}

	return result
}

// 将数据流转换为切片
func takeN(done <-chan struct{}, valueStream <-chan interface{}, num int) []interface{} {
	taken := make([]interface{}, 0, num)
	for i := 0; i < num; i++ {
		select {
		case <-done:
			return taken
		case val, ok := <-valueStream:
			if !ok {
				return taken
			}
			taken = append(taken, val)
		}
	}

	return taken
}

func main() {
	// 设置随机数种子
	rand.Seed(time.Now().UnixNano())

	// 创建done channel以支持取消
	done := make(chan struct{})
	defer close(done)

	fmt.Println("生成一系列channels，每个channel包含5个整数")
	fmt.Println("桥接将所有这些channel扁平化为一个stream")

	// 创建一个channels流
	chanStream := genChannelStream(done)

	// 将channels流桥接成一个平坦的值流
	valueStream := bridge(done, chanStream)

	// 使用函数式操作处理桥接后的流

	// 1. 使用map将值翻倍
	doubledStream := mapStream(done, valueStream, func(v interface{}) interface{} {
		n := v.(int)
		return n * 2
	})

	// 2. 使用filter只保留能被3整除的值
	filteredStream := filterStream(done, doubledStream, func(v interface{}) bool {
		n := v.(int)
		return n%3 == 0
	})

	// 创建一个计数和处理所有输出的goroutine
	fmt.Println("处理桥接、映射和过滤后的结果流:")

	var wg sync.WaitGroup
	wg.Add(1)

	go func() {
		defer wg.Done()

		// 3. 使用reduce计算总和
		sum := reduceStream(done, filteredStream, func(acc, val interface{}) interface{} {
			accumulated := acc.(int)
			n := val.(int)

			fmt.Printf("值: %d (累积总和: %d)\n", n, accumulated+n)
			return accumulated + n
		}, 0)

		fmt.Printf("最终总和: %d\n", sum)
	}()

	// 等待处理完成
	wg.Wait()

	// 展示takeN函数的使用
	fmt.Println("=== 另一个桥接示例与takeN ===")

	// 创建新的channels流
	chanStream2 := genChannelStream(done)
	valueStream2 := bridge(done, chanStream2)

	// 只获取前10个值
	fmt.Println("仅获取前10个值:")
	values := takeN(done, valueStream2, 10)

	fmt.Println("收集的值:")
	for i, v := range values {
		fmt.Printf("值 %d: %v\n", i+1, v)
	}

	// 提前取消示例
	fmt.Println("=== 提前取消的桥接示例 ===")

	earlyDone := make(chan struct{})

	// 创建新的channels流
	chanStream3 := genChannelStream(earlyDone)
	valueStream3 := bridge(earlyDone, chanStream3)

	// 获取前5个值后取消
	fmt.Println("获取5个值后取消:")

	count := 0
	for val := range valueStream3 {
		fmt.Printf("接收到: %v\n", val)
		count++

		if count >= 5 {
			fmt.Println("已收到5个值，取消操作...")
			close(earlyDone)
			break
		}
	}

	// 给goroutines时间清理
	time.Sleep(time.Second)
	fmt.Println("所有操作完成")
}
